package com.yyfax.account.service;

import com.google.common.base.Function;
import com.google.common.collect.ImmutableListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimaps;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.jaws.core.common.log.Log;
import com.jaws.core.common.util.DateUtil;
import com.yyfax.account.common.enums.DayChangeStateEnum;
import com.yyfax.account.common.enums.TableNameEnum;
import com.yyfax.account.common.exception.AccountExceptionCode;
import com.yyfax.account.common.util.TableSegmentUtils;
import com.yyfax.account.dao.AccountFrozenDao;
import com.yyfax.account.dao.AccountInfoDao;
import com.yyfax.account.dao.AccountTradeDao;
import com.yyfax.account.dao.AccountTradeFrozenDao;
import com.yyfax.account.dao.DayChangeRecordDao;
import com.yyfax.account.dao.DayChangeResultDao;
import com.yyfax.account.domain.DayChangeRecord;
import com.yyfax.account.domain.DayChangeResult;
import com.yyfax.account.vo.TradeSummary;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 日切
 *
 * @author maojian
 * @date 2017-10-05 09:38:03
 * @since v1.0.0
 */
@Service
public class DayChangeTransService {
	private static DateTimeFormatter formatter = DateTimeFormat.forPattern(DateUtil.FORMAT_DATE);
	private final static int DEFAULT_THREAD_NUM = (Runtime.getRuntime().availableProcessors() + 1 > 5) ? 6
			: Runtime.getRuntime().availableProcessors() + 1;
	private Logger logger = LoggerFactory.getLogger(getClass());
	@Resource
	private AccountTradeDao accountTradeDao;
	@Resource
	private AccountFrozenDao accountFrozenDao;
	@Resource
	private AccountTradeFrozenDao accountTradeFrozenDao;
	@Resource
	private DayChangeResultDao dayChangeResultDao;
	@Resource
	private DayChangeRecordDao dayChangeRecordDao;
	@Resource
	private AccountInfoDao accountInfoDao;

	@Value("${DAY_CHANGE_BATCH_HANDLE_SIZE:3500}")
	private Integer DAY_CHANGE_BATCH_HANDLE_SIZE;

	@Value("${DAY_CHANGE_BATCH_INSERT_SIZE:1000}")
	private Integer DAY_CHANGE_BATCH_INSERT_SIZE;

	enum UniqueDayChangeResultUserId implements Function<DayChangeResult, String> {
		/**
		 * 实例
		 */
		INSTANCE;

		@Override
		public String apply(DayChangeResult input) {
			return input.getUserId();
		}
	}

	enum UniqueTradeSummaryUserId implements Function<TradeSummary, String> {
		/**
		 * 实例
		 */
		INSTANCE;

		@Override
		public String apply(TradeSummary input) {
			return input.getUserId();
		}
	}

	enum AccountTradeSegment implements Function<String, String> {
		/**
		 * 实例
		 */
		INSTANCE;

		@Override
		public String apply(String userId) {
			return TableSegmentUtils.getTableSegment(TableNameEnum.ACCOUNT_TRADE, userId);
		}
	}

	enum AccountFrozenSegment implements Function<String, String> {
		/**
		 * 实例
		 */
		INSTANCE;

		@Override
		public String apply(String userId) {
			return TableSegmentUtils.getTableSegment(TableNameEnum.ACCOUNT_FROZEN, userId);
		}
	}

	enum AccountComplexTradeSegment implements Function<String, String> {
		/**
		 * 实例
		 */
		INSTANCE;

		@Override
		public String apply(String userId) {
			return TableSegmentUtils.getTableSegment(TableNameEnum.ACCOUNT_TRADE_FROZEN, userId);
		}
	}

	/**
	 * 构建制定日期日切结果
	 *
	 * @param preDayChangeResult 指定日切日期前一日 日切结果
	 * @param changeDay          日切日期
	 * @return DayChangeResult 指定日期日切结果
	 */
	private DayChangeResult buildDayChangeResult(DayChangeResult preDayChangeResult, DateTime changeDay) {
		DayChangeResult result = new DayChangeResult();

		result.setUserId(preDayChangeResult.getUserId());
		result.setAccountNo(preDayChangeResult.getAccountNo());
		result.setAccountName(preDayChangeResult.getAccountName());
		result.setAccountAliasName(preDayChangeResult.getAccountAliasName());
		result.setDepositAccountNo(preDayChangeResult.getDepositAccountNo());
		result.setAccountType(preDayChangeResult.getAccountType());
		result.setCreateTime(DateUtil.getCurrentDateTime());
		result.setChangeDay(formatter.print(changeDay));

		return result;
	}

	/**
	 * 执行日切数据处
	 *
	 * @param appointDate 日切日期
	 * @param record      日切记录
	 * @author liangqm
	 * @date 2017-10-16 13:40:12
	 * @since v1.0.0
	 */
	void doDayChangeByDate(DateTime appointDate, DayChangeRecord record) {
		final String op = "DayChangeTransService.doDayChangeByDate";

		// 创建分页对象,分批处理账户交易数据
		Batch batch = new Batch(formatter.print(appointDate));

		// 创建线程池
		ThreadFactory nameThreadFactory = new ThreadFactoryBuilder().setNameFormat("day_change-%d").build();
		ExecutorService executorService = new ThreadPoolExecutor(DEFAULT_THREAD_NUM, DEFAULT_THREAD_NUM, 0L, TimeUnit.MILLISECONDS,
				new LinkedBlockingQueue<Runnable>(), nameThreadFactory);

		final CountDownLatch countDownLatch = new CountDownLatch(batch.getBatchNum());

		// 多线程计算日切数据
		for (int batchNo = 0; batchNo < batch.getBatchNum(); batchNo++) {
			executorService.execute(new DayChangeInternalService(appointDate, batchNo, batch, countDownLatch, record));
		}

		try {
			countDownLatch.await();
		} catch (Exception e) {
			logger.error(Log.op(op).msg("批量处理日切数据异常").toString(), e);
			record.setState(DayChangeStateEnum.FAIL.getValue());
			dayChangeRecordDao.update(record);
		} finally {
			executorService.shutdown();// 关闭线程池
		}
	}

	/**
	 * 日切服务内部类
	 * 主要负责分批计算日切数据、入库
	 * 本类只通过线程池调用,run()为执行入口
	 * 核心逻辑在handleDayChangeByPage()
	 *
	 * @author liangqm
	 * @date 2017-10-25 13:57:20
	 * @since v1.0.0
	 */
	class DayChangeInternalService implements Runnable {
		/**
		 * 批次号
		 */
		private Integer batchNo;

		/**
		 * 分批信息
		 */
		private Batch batch;

		/**
		 * 日切日期
		 */
		private DateTime appointDate;

		/**
		 * 日切日期前一天
		 */
		private String preDate;

		/**
		 * 线程计数器
		 */
		private CountDownLatch countDownLatch;

		/**
		 * 日切记录
		 */
		private DayChangeRecord record;

		DayChangeInternalService(DateTime appointDate, Integer batchNo, Batch page, CountDownLatch countDownLatch, DayChangeRecord record) {
			this.appointDate = appointDate;
			this.preDate = formatter.print(appointDate.minusDays(1));
			this.batchNo = batchNo;
			this.batch = page;
			this.countDownLatch = countDownLatch;
			this.record = record;
		}

		/**
		 * 处理日切记录
		 * 1、获取当前日切任务前一日的日切记录
		 * 2、获取当前日切日期账户交易（account_trad）和冻结(account_frozen)总和
		 * 3、计算当前日切任务的日切记录：期初= 前一日日切期末; 交易（冻结） 增加 = account_trad（account_frozen）.sum(amount); 期末= 当前期初+ 今日新增- 今日减少交易
		 *
		 * @author liangqm
		 * @date 2017-10-16 13:40:12
		 * @since v1.0.0
		 */
		private void handleDayChangeByPage() {
			final String op = "DayChangeTransService.DayChangeInternalService.handleDayChangeByPage";
			logger.info(Log.op(op).kv("pageNo", this.batchNo).msg("分批处理日切数据--begin").toString());
			Long start = System.currentTimeMillis();
			// 1. 查询前一天日切数据 account_info left join account_day_change_result
			List<DayChangeResult> preDayChangeResults = queryPreDayChangeResults();
			logger.info(Log.op(op).msg("queryPreDayChangeResults 耗时" + (System.currentTimeMillis() - start)).toString());

			// key = userId , value = 前一日日切结果
			Map<String, DayChangeResult> accountDayChangeMap = Maps.uniqueIndex(preDayChangeResults, UniqueDayChangeResultUserId.INSTANCE);
			// userId
			List<String> userIdList = Lists.transform(preDayChangeResults, UniqueDayChangeResultUserId.INSTANCE);

			// 2. 获取当前日切日期账户交易（account_trad）和冻结(account_frozen)总和
			// 2.1 获取每个账户今日交易和支出总和 key = userId
			Map<String, TradeSummary> tradeSumMap = obtainTradeSumMap(userIdList);
			// 2.2 获取每个账户今日冻结和解冻总和 key = userId
			Map<String, TradeSummary> frozenSumMap = obtainFrozenSumMap(userIdList);
			// 2.3 获取每个账户今日冻结增加余额 和 解冻扣减余额总和 key = userId
			Map<String, TradeSummary> complexTradeSumMap = obtainComplexTradeSumMap(userIdList);


			// 3. 前一日日切期末、交易数据 、冻结数据 汇总成当日日切结果
			mergeDayChangeResult(accountDayChangeMap, tradeSumMap, frozenSumMap, complexTradeSumMap);

			logger.info(Log.op(op).kv("pageNo", this.batchNo).msg("分批处理日切数据--end").toString());
		}


		/**
		 * 获取当前日切任务前一日日切记录
		 * 如果任务为REDOING状态,排除当前日切任务已经入库的记录
		 *
		 * @author liangqm
		 * @date 2017-10-16 13:40:12
		 * @since v1.0.0
		 */
		private List<DayChangeResult> queryPreDayChangeResults() {
			List<DayChangeResult> preDayChangeResults = Lists.newArrayList();

			// 当前执行的日切任务不能为空
			if (record == null) {
				throw AccountExceptionCode.DAY_CHANGE_RECORD_NOT_NULL.exp(formatter.print(appointDate));
			}

			if (DayChangeStateEnum.REDOING.isValue(record.getState())) {
				// 日切任务重新执行，分批查询前一日日切记录，排除之前执已经入库的记录
				preDayChangeResults = dayChangeResultDao.batchQueryPreDayChangeResultExcludeExist(batch.batchNoMap.get(batchNo),
						batch.getBatchHandleSize(), preDate, formatter.print(appointDate));

			} else if (DayChangeStateEnum.DEALING.isValue(record.getState())) {
				// 日切任务是正在执行状态,分批查询前一日日切记录
				preDayChangeResults = dayChangeResultDao.batchQueryPreDayChangeResult(batch.batchNoMap.get(batchNo), batch.getBatchHandleSize(),
						preDate, formatter.print(appointDate));
			}
			return preDayChangeResults;
		}

		/**
		 * 合并 前一日日切期末、交易数据 、冻结数据 汇总成当日日切结果
		 *
		 * @param accountDayChangeMap 前一日账户日切结果
		 * @param tradeSummaryMap     前一日账户交易sum结果
		 * @param frozenInfoMap       前一日账户冻结sum结果
		 * @param complexTradeSumMap
		 * @author liangqm
		 * @date 2017-10-16 13:40:12
		 * @since v1.0.0
		 */
		private void mergeDayChangeResult(Map<String, DayChangeResult> accountDayChangeMap, Map<String, TradeSummary> tradeSummaryMap,
										  Map<String, TradeSummary> frozenInfoMap, Map<String, TradeSummary> complexTradeSumMap) {
			final String op = "DayChangeTransService.DayChangeInternalService.mergeDayChangeResult";

			Iterator<Map.Entry<String, DayChangeResult>> iterator = accountDayChangeMap.entrySet().iterator();
			List<DayChangeResult> dayChangeResults = Lists.newArrayList();

			while (iterator.hasNext()) {
				Map.Entry<String, DayChangeResult> entry = iterator.next();
				String userId = entry.getKey();
				DayChangeResult preDayChangeResult = entry.getValue();

				// 构建新的日切结果对象
				DayChangeResult nowResult = buildDayChangeResult(preDayChangeResult, appointDate);

				// 获取日切交易数据 期初、新增、减少、期末 (单位:分)
				obtainTradeDayChange(userId, preDayChangeResult, tradeSummaryMap, complexTradeSumMap, nowResult);

				// 获取日切冻结数据 期初、新增、减少、期末 (单位:分)
				obtainFrozenDayChange(userId, preDayChangeResult, frozenInfoMap, complexTradeSumMap, nowResult);

				// 日切对象添加到集合中，执行批量插入
				if (nowResult.getAmount() != null && nowResult.getAmount() > 0) {
					dayChangeResults.add(nowResult);
				}
			}

			try {
				Long start = System.currentTimeMillis();
				// 批量插入
				if (dayChangeResults.size() > 0) {
					if (dayChangeResults.size() > DAY_CHANGE_BATCH_INSERT_SIZE) {
						List<List<DayChangeResult>> dayChangeResultPage = Lists.partition(dayChangeResults, DAY_CHANGE_BATCH_INSERT_SIZE);

						for (List<DayChangeResult> dayChangeResultList : dayChangeResultPage) {
							dayChangeResultDao.insertBatch(dayChangeResultList);
						}
					} else {
						dayChangeResultDao.insertBatch(dayChangeResults);
					}
				}
				logger.info(Log.op("DayChangeInternalService.insertBatch").msg(" insertBatch 耗时 " + (System.currentTimeMillis() - start)).toString());
			} catch (DuplicateKeyException e) {
				logger.error(Log.op(op).kv("Batch", batch.batchNoMap).kv("batchNo", batchNo).msg("批量插入日切记录失败").toString(), e);
				throw AccountExceptionCode.DAY_CHANGE_INSERT_ERROR.exp();
			}
		}

		/**
		 * 获取冻结日切数据 期初、新增、减少、期末 (单位:分)
		 *
		 * @param userId             用户id
		 * @param preDayChangeResult 前一日日切结果
		 * @param frozenInfoMap      账户前一日冻结sum结果
		 * @param nowResult          当日日切数据
		 * @author liangqm
		 * @date 2017-10-16 13:40:12
		 * @since v1.0.0
		 */
		private void obtainFrozenDayChange(String userId, DayChangeResult preDayChangeResult, Map<String, TradeSummary> frozenInfoMap, Map<String, TradeSummary> complexTradeSumMap, DayChangeResult nowResult) {
			long preFrozenAmount;
			long frozenIncreaseAmount = 0L;
			long frozenDecreaseAmount = 0L;
			TradeSummary frozenSummary = frozenInfoMap.get(userId);
			TradeSummary complexSummary = complexTradeSumMap.get(userId);

			// 冻结 新增 减少
			if (null != frozenSummary) {
				frozenIncreaseAmount = frozenSummary.getIncreaseAmount();
				frozenDecreaseAmount = frozenSummary.getDecreaseAmount();
			}

			// 复合交易 冻结数据
			if (null != complexSummary) {
				frozenIncreaseAmount = frozenIncreaseAmount + complexSummary.getIncreaseAmount();
				frozenDecreaseAmount = frozenDecreaseAmount + complexSummary.getDecreaseAmount();
			}

			// 冻结期初余额
			if (preDayChangeResult.getFrozenAmount() == null) {
				preFrozenAmount = 0;
			} else {// 冻结期初余额 = 前一日冻结期末余额
				preFrozenAmount = preDayChangeResult.getFrozenAmount();
			}

			// 冻结期初
			nowResult.setPreFrozenAmount(preFrozenAmount);
			// 冻结新增
			nowResult.setFrozenIncreaseAmount(frozenIncreaseAmount);
			// 冻结减少
			nowResult.setFrozenDecreaseAmount(frozenDecreaseAmount);
			// 冻结期末
			nowResult.setFrozenAmount(nowResult.getPreFrozenAmount() + nowResult.getFrozenIncreaseAmount() - nowResult.getFrozenDecreaseAmount());
		}


		/**
		 * 查询账户(日切日期)前一日复合交易(冻结增加余额 和 解冻扣减余额) 数据
		 * 1.根据userid 获取分表情况,account_trade_frozen{segment} 对应哪些userId
		 * 2.循环account_trade_frozen{segment} ，sum 账户今日冻结和解冻金额
		 *
		 * @param userIdList 用户id
		 * @return Map key = userId ; TradeSummary对象包含 冻结 和 解冻金额
		 * @author liangqm
		 * @date 2017-10-16 13:40:12
		 * @since v1.0.0
		 */
		private Map<String, TradeSummary> obtainComplexTradeSumMap(List<String> userIdList) {
			Map<String, TradeSummary> accountComplexTradeMap = Maps.newLinkedHashMap();
			// 1、根据userId计算 account_trade 分表
			// Multimaps.index 把List转换成一键多值Map(key 可以重复的map), key = table_segment value = userId
			ImmutableListMultimap<String, String> complexTradeSegmentMap = Multimaps.index(userIdList, AccountComplexTradeSegment.INSTANCE);

			// 2、循环分表,计算账户 冻结 解冻 金额
			int i = 0;
			for (String complexTradeSegment : complexTradeSegmentMap.keySet()) {
				Set<String> userIdSet = Sets.newHashSet(complexTradeSegmentMap.get(complexTradeSegment));
				Long start = System.currentTimeMillis();
				List<TradeSummary> complexTradeSummaryList = accountTradeFrozenDao.queryComplexTradeSumBatchByDate(userIdSet, preDate, formatter.print(appointDate), complexTradeSegment);
				logger.info(Log.op("DayChangeInternalService.queryComplexTradeSumBatchByDate").msg(i + " queryComplexTradeSumBatchByDate 耗时 " + (System.currentTimeMillis() - start)).toString());
				Map<String, TradeSummary> tempMap = Maps.uniqueIndex(complexTradeSummaryList, UniqueTradeSummaryUserId.INSTANCE);
				accountComplexTradeMap.putAll(tempMap);
				++i;
			}

			return accountComplexTradeMap;
		}

		/**
		 * 查询账户(日切日期)前一日冻结数据
		 * 1.根据userid 获取分表情况,account_frozen{segment} 对应哪些userId
		 * 2.循环account_frozen{segment} ，sum 账户今日冻结和解冻金额
		 *
		 * @param userIdList 用户id
		 * @return Map key = userId ; TradeSummary对象包含 冻结 和 解冻金额
		 * @author liangqm
		 * @date 2017-10-16 13:40:12
		 * @since v1.0.0
		 */
		private Map<String, TradeSummary> obtainFrozenSumMap(List<String> userIdList) {
			Map<String, TradeSummary> accountFrozenMap = Maps.newLinkedHashMap();

			// 1、根据userId计算 account_trade 分表
			// Multimaps.index 把List转换成一键多值Map(key 可以重复的map), key = table_segment value = userId
			ImmutableListMultimap<String, String> frozenSegmentMap = Multimaps.index(userIdList, AccountFrozenSegment.INSTANCE);

			// 2、循环分表,计算账户 冻结 解冻 金额
			int i = 0;
			for (String frozenSegment : frozenSegmentMap.keySet()) {
				Set<String> userIdSet = Sets.newHashSet(frozenSegmentMap.get(frozenSegment));
				Long start = System.currentTimeMillis();
				List<TradeSummary> frozenSummaryList = accountFrozenDao.queryFrozenSumBatchByDate(userIdSet, preDate, formatter.print(appointDate),
						frozenSegment);
				logger.info(Log.op("DayChangeInternalService.queryFrozenSumBatchByDate").msg(i + " queryFrozenSumBatchByDate 耗时 " + (System.currentTimeMillis() - start)).toString());
				Map<String, TradeSummary> tempMap = Maps.uniqueIndex(frozenSummaryList, UniqueTradeSummaryUserId.INSTANCE);
				accountFrozenMap.putAll(tempMap);

				++i;
			}
			return accountFrozenMap;
		}

		/**
		 * 获取账户交易日切数据 期初、新增、减少、期末 (单位:分)
		 *
		 * @param userId             用户id
		 * @param preDayChangeResult 前一日日切结果
		 * @param complexTradeSumMap 复合交易sum结果
		 * @param tradeSummaryMap    账户前一日交易sum结果
		 * @param nowResult          当日日切结果
		 * @author liangqm
		 * @date 2017-10-16 13:40:12
		 * @since v1.0.0
		 */
		private void obtainTradeDayChange(String userId, DayChangeResult preDayChangeResult, Map<String, TradeSummary> tradeSummaryMap, Map<String, TradeSummary> complexTradeSumMap, DayChangeResult nowResult) {
			long preAmount;
			long increaseAmount = 0L;
			long decreaseAmount = 0L;
			TradeSummary tradeSummary = tradeSummaryMap.get(userId);
			TradeSummary complexSummary = complexTradeSumMap.get(userId);

			// 账面余额 新增 减少
			// 从前一日交易数据获取账面余额增加、减少
			if (null != tradeSummary) {
				increaseAmount = tradeSummary.getIncreaseAmount();
				decreaseAmount = tradeSummary.getDecreaseAmount();
			}

			// 复合交易记录
			if (null != complexSummary) {
				increaseAmount = increaseAmount + complexSummary.getIncreaseAmount();
				decreaseAmount = decreaseAmount + complexSummary.getDecreaseAmount();
			}

			// 期初余额
			if (preDayChangeResult.getAmount() == null) {
				preAmount = 0;
			} else {
				// 前一日日切期末 = 当日日切期初
				preAmount = preDayChangeResult.getAmount();
			}

			// 期初
			nowResult.setPreAmount(preAmount);
			// 新增
			nowResult.setIncreaseAmount(increaseAmount);
			// 减少
			nowResult.setDecreaseAmount(decreaseAmount);
			// 期末(期初 + 新增 - 减少)
			nowResult.setAmount(nowResult.getPreAmount() + nowResult.getIncreaseAmount() - nowResult.getDecreaseAmount());
		}

		/**
		 * 获取用户（日切日期）前一日交易数据
		 * 1.根据userid 获取分表情况,account_trade{segment} 对应哪些userId
		 * 2.循环account_trade{segment} ，sum 账户今日交易和支出金额
		 *
		 * @param userIdList 用户id
		 * @return Map key = userId ; TradeSummary对象包含 增加和减少金额
		 * @author liangqm
		 * @date 2017-10-16 13:40:12
		 * @since v1.0.0
		 */
		private Map<String, TradeSummary> obtainTradeSumMap(List<String> userIdList) {
			Map<String, TradeSummary> accountTradeMap = Maps.newLinkedHashMap();

			// 1、根据userId计算 account_trade 分表
			// Multimaps.index 把List转换成一键多值Map(key 可以重复的map), key = table_segment value = userId
			ImmutableListMultimap<String, String> tradeSegmentMap = Multimaps.index(userIdList, AccountTradeSegment.INSTANCE);

			// 2、循环分表,计算账户交易 增加 减少金额
			int i = 0;
			for (String tradeSegment : tradeSegmentMap.keySet()) {
				List<String> segmentUserList = tradeSegmentMap.get(tradeSegment);
				Long start = System.currentTimeMillis();
				List<TradeSummary> tradeSummaryList = accountTradeDao.batchQueryTradeSumByDate(segmentUserList, preDate, formatter.print(appointDate),
						tradeSegment);
				logger.info(Log.op("DayChangeInternalService.obtainTradeSumMap").msg(i + " batchQueryTradeSumByDate 耗时 " + (System.currentTimeMillis() - start)).toString());
				Map<String, TradeSummary> tempMap = Maps.uniqueIndex(tradeSummaryList, UniqueTradeSummaryUserId.INSTANCE);

				accountTradeMap.putAll(tempMap);
				++i;
			}

			return accountTradeMap;
		}

		@Override
		public void run() {
			final String op = "DayChangeInternalService.run";

			try {
				handleDayChangeByPage();
			} catch (Exception e) {
				logger.error(Log.op(op).kv("pageNo", this.batchNo).kv("page", this.batch.batchNoMap).msg("处理日切数据出现异常").toString(), e);
				// 多线程并发处理，当一个线程内部出现异常，更新日切任务为FAIL，不打断其他线程执行
				record.setState(DayChangeStateEnum.FAIL.getValue());
				dayChangeRecordDao.update(record);
			} finally {
				countDownLatch.countDown();
				MDC.clear();
			}
		}
	}

	/**
	 * 分批信息里类，只供DayChangeTransService 使用
	 * 包括：总记录数，批次数量，每批查询数量,批次与起始id对应关系
	 *
	 * @author liangqm
	 * @date 2017-10-25 14:49:18
	 * @since v1.0.0
	 */
	class Batch {

		/**
		 * 日切任务日期
		 */
		String dayChangeRecordDate;

		/**
		 * 总记录数
		 */
		Integer totalSize;

		/**
		 * 批次数量
		 */
		Integer batchNum;

		/**
		 * 每批处理数量
		 */
		final Integer batchHandleSize = DAY_CHANGE_BATCH_HANDLE_SIZE;

		/**
		 * key = 批次号 value= 查询起始id号
		 */
		Map<Integer, Long> batchNoMap;

		Batch(String dayChangeRecordDate) {
			this.dayChangeRecordDate = dayChangeRecordDate;
			// 查询参与日切的账户总数
			this.totalSize = accountInfoDao.queryAccountRecordCount(dayChangeRecordDate);

			// 总数为0 页数也为0
			if (totalSize == 0) {
				batchNum = 0;
			} else {
				batchNum = (totalSize - 1) / batchHandleSize + 1;
			}

			// 计算每页起始id
			batchNoMap = generateBatchNoMap();
		}

		/**
		 * 大数据量分页使用id > appointID LIMIT pageSize 进行分页
		 * 由于使用多线程处理，所以需要提前计算每一页起始account_info.id
		 *
		 * @return Map key=页数 value=起始id
		 * @author liangqm
		 * @date 2017-10-25 13:49:18
		 * @since v1.0.0
		 */
		private Map<Integer, Long> generateBatchNoMap() {
			Map<Integer, Long> map = Maps.newLinkedHashMap();
			Long preMaxId = 0L;

			for (int i = 0; i < this.batchNum; i++) {
				if (i == 0) {
					map.put(0, 0L);
					continue;
				}

				preMaxId = accountInfoDao.queryAccountPageMap(dayChangeRecordDate, preMaxId, this.batchHandleSize);
				map.put(i, preMaxId);
			}

			logger.info(Log.op("Batch.generateBatchNoMap").kv("batch", map).msg("日切分批处理信息").toString());
			return map;
		}

		Integer getBatchNum() {
			return batchNum;
		}

		Integer getBatchHandleSize() {
			return batchHandleSize;
		}
	}
}